package com.atbeijing.D09;

import com.atbeijing.D04.Example4;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Slide;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.Timestamp;
import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * 事件时间,开窗
 * 计算每个商品的pv数量
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Example4.UserBehavior> stream = env
                .readTextFile("C:\\Users\\WangJinGen\\IdeaProjects\\flink\\data\\UserBehavior.csv")
                .map(new MapFunction<String, Example4.UserBehavior>() {
                    @Override
                    public Example4.UserBehavior map(String value) throws Exception {
                        String[] arr = value.split(",");
                        return new Example4.UserBehavior(
                                arr[0], arr[1], arr[2], arr[3], Long.parseLong(arr[4]) * 1000L
                        );
                    }
                })
                .filter(r -> r.behaviorType.equals("pv"))
                //设置水位线
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //延迟时间,这里已经知道数据是按照时间戳升序来的
                                .<Example4.UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                //水位线参照字段
                                .withTimestampAssigner(new SerializableTimestampAssigner<Example4.UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(Example4.UserBehavior element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })

                );

        //配置表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();

        //获取表环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        tEnv.createTemporaryView(
                "userbehavior",
                stream, $("itemId"),
                $("timestamp").rowtime().as("ts") // 将timestamp字段设置为事件时间
        );

        // HOP 滑动窗口，5分钟的滑动距离，1小时的窗口长度
        Table sqlResult = tEnv.sqlQuery("SELECT itemId, COUNT(itemId), HOP_END(ts, INTERVAL '5' MINUTE, INTERVAL '1' HOUR) FROM userbehavior GROUP BY itemId, HOP(ts, INTERVAL '5' MINUTE, INTERVAL '1' HOUR)");

        tEnv.toRetractStream(sqlResult, Row.class).print();

        Table table = tEnv.fromDataStream(
                stream, $("itemId"),
                $("timestamp").rowtime().as("ts") // 将timestamp字段设置为事件时间
        );

        Table tableResult = table
                .window(Slide.over(lit(1).hours()).every(lit(5).minutes()).on($("ts")).as("w"))
                .groupBy($("itemId"), $("w"))
                .select($("itemId"), $("itemId").count(), $("w").end());

        tEnv.toRetractStream(tableResult, Row.class).print();

        env.execute();
    }

    public static class UserBehavior {
        public String userId;
        public String itemId;
        public String categoryId;
        public String behaviorType;
        public Long timestamp;

        public UserBehavior() {
        }

        public UserBehavior(String userId, String itemId, String categoryId, String behaviorType, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.categoryId = categoryId;
            this.behaviorType = behaviorType;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "UserBehavior{" +
                    "userId='" + userId + '\'' +
                    ", itemId='" + itemId + '\'' +
                    ", categoryId='" + categoryId + '\'' +
                    ", behaviorType='" + behaviorType + '\'' +
                    ", timestamp=" + new Timestamp(timestamp) +
                    '}';
        }
    }

}
